package com.fzk.core.btree;

import com.fzk.log.Logger;
import lombok.Data;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import java.nio.charset.StandardCharsets;

/**
 * 基于文件的B+树，实现了LRU缓存控制
 *
 * @author fzk
 * @datetime 2023-05-12 23:24:26
 */
public class Page {
    private static final int BinarySearchThreshold = 8;// 二分查找阈值，数组较小时遍历平均情况下会比较次数更少
    // 将0作为非法页下标，这是因为文件中默认值都是0，如果pageIdx从0开始，会造成含义不清
    public static final int Invalid_PageIdx = 0;// 非法页下标, 注意缓存页的页下标为0
    public static final int Index_Page_Type = 1;// 索引页
    public static final int Leaf_Page_Type = 2;// 叶子页
    @SuppressWarnings("unused")
    public static final int Tmp_Page_Type = 3;// 缓存页，暂时未用

    public static final int Page_Size = 4 * 1024;// 默认一页4KB
    // 一页内包含的key数量有2个限制：
    // pageUsedSize<=Page_size 且 keyCount<=Key_Max_Num
    public static final int Key_Max_Num = 8;// 页内最多包含key数量

    // ----------------------page header----------------
    public int pageType;// 页类型: 1为索引页, 2为叶子页
    /**
     * 本页下标，从1开始使用，0是非法值，
     * 这是因为文件中默认值都是0，如果pageIdx从0开始，会造成含义不清
     */
    public int pageIdx;
    public int parentPageIdx;// 父页面下标

    public static void directWriteParentPageIdx(FileChannel fileChannel, long pagePos, int parentPageIdx) throws IOException {
        absWritePageHeader(fileChannel, pagePos, parentPageIdx, 4 * 2);
    }

    public int keyCount;// key的数量
    public int prePageIdx = Invalid_PageIdx;// 索引页面和页子页面都维护双向链表

    public static void directWritePrePageIdx(FileChannel fileChannel, long pagePos, int prePageIdx) throws IOException {
        absWritePageHeader(fileChannel, pagePos, prePageIdx, 4 * 4);
    }

    public int nextPageIdx = Invalid_PageIdx;// 索引页面和页子页面都维护双向链表

    public static void directWriteNextPageIdx(FileChannel fileChannel, long pagePos, int nextPageIdx) throws IOException {
        absWritePageHeader(fileChannel, pagePos, nextPageIdx, 4 * 5);
    }

    public int pageUsedSize = Page_Header_Meta_Size;// 页面已用字节数

    /**
     * 直接在文件中更新此页的某个页头
     * 注意：如果此页已经被读取到内存，缓存中的页无法感知文件里的内容已变化，这个问题需要注意
     * 因此这个方法只能在缓存中不存在此页时才可以使用, 目的在于避免读取页到缓存中
     */
    private static void absWritePageHeader(FileChannel fileChannel, long pagePos, int val, int valOff) throws IOException {
        ByteBuffer buf = ByteBuffer.allocate(4);
        buf.putInt(val);
        buf.flip();// 翻转待读
        fileChannel.write(buf, pagePos + valOff);
    }

    public static final int Page_Header_Meta_Size = 4 * 7;// 页头大小，页元信息
    // ----------------------page header----------------

    public BEntry[] entries;
    public volatile boolean modified = false;// 标记此页面是否发生修改
    public volatile boolean isInCache = false;// 标记是否位于lru缓存中
    public volatile boolean deleted = false;// 是否已经从B+树中删除了

    public Page(int pageType, int pageIdx, int parentPageIdx) {
        this.pageType = pageType;
        this.pageIdx = pageIdx;
        this.parentPageIdx = parentPageIdx;
        this.keyCount = 0;
        this.entries = new BEntry[Key_Max_Num];
    }

    /**
     * 查找key的索引, 如果不存在则返回其应该插入的索引
     *
     * @return 返回存在时的索引或不存在时应该插入的索引
     */
    public SearchResult searchKeyIndex(String key) {
        if (keyCount <= 0) return new SearchResult(false, 0);
        // 二分查找阈值，数组较小时遍历平均情况下会比较次数更少
        if (keyCount >= BinarySearchThreshold) {
            int left = 0, right = keyCount - 1;
            while (left <= right) {
                int mid = (left + right) >> 1;
                int cmp = key.compareTo(entries[mid].key);
                if (cmp < 0) right = mid - 1;
                else if (cmp == 0) return new SearchResult(true, mid);
                else left = mid + 1;
            }
            return new SearchResult(false, left);
        } else {
            // 遍历
            for (int i = 0; i < keyCount; i++) {
                int cmp = key.compareTo(entries[i].key);
                if (cmp == 0) {
                    return new SearchResult(true, i);
                } else if (cmp < 0) {// 找到了第一个比key大的作为插入索引
                    return new SearchResult(false, i);
                }
            }
            // 都比key小，则最右边为插入索引
            return new SearchResult(false, keyCount);
        }
    }

    // 不存在返回null
    public Long get(String key) {
        SearchResult result = searchKeyIndex(key);
        if (result.exists) {
            return entries[result.index].val;
        }
        return null;
    }

    /**
     * 添加到页内, 存在相同key时覆盖
     *
     * @param key 插入的key
     * @param val 子页面页下标或文件存储元信息offset
     * @return 存在旧值返回，否则null
     */
    public Long put(String key, long val, BTreeFile btree) throws IOException {
        // 1.找到插入的索引位置
        SearchResult result = searchKeyIndex(key);
        Long old = null;
        boolean updateMinKeyFlag = false;
        if (result.exists) {
            // 存在旧值，直接更新
            old = this.entries[result.index].val;
            this.entries[result.index].val = val;
            this.modified = true;// 标记页面已经修改
            Logger.debug(String.format("更新key: %s, oldVal: %d, newVal: %d", key, old, val));
        } else {
            // 新key，插入
            byte[] data = key.getBytes(StandardCharsets.UTF_8);
            int keyLen = data.length;
            int index = result.index;// 待插入索引
            // 2.检查
            if (!canInsert(key)) {
                throw new RuntimeException(String.format(
                        "向page插入key时检查错误, key槽写满了或页面空间不足, keyCount: %d, pageUsedSize: %d, key数据长度: %d",
                        keyCount, pageUsedSize, keyLen));
            }

            // 若minKey更新需要维护父节点指向此节点的key,
            // 若keyCount为0说明是空白页, 此时父子结点关系还未建立，父节点指向子结点的key不在这里维护
            if (index == 0 && keyCount > 0) {
                updateMinKeyFlag = true;
            }

            // 移动并插入新key
            BEntry entry = new BEntry(key, val);
            System.arraycopy(entries, index, entries, index + 1, keyCount - index);
            entries[index] = entry;

            // 更新页头的key数量和已使用空间
            keyCount++;
            increasePageUsedSize(data);
            this.modified = true;// 标记页面已经修改
        }

        // 更新minKey
        if (updateMinKeyFlag) {
            String oldMinKey = entries[1].key;
            btree.minKeyUpdated(this, oldMinKey, key);
        }
        return old;
    }

    /**
     * 替换页面的key, 一般用于minKey更新
     * 若替换的key为minKey则返回0可用于递归操作minKey更新
     *
     * @param oldKey 旧key，必须存在否则报错
     * @param newKey 替换的新key
     * @return 替换的oldKey在数组的索引
     */
    public int replaceKey(String oldKey, String newKey) {
        SearchResult result = searchKeyIndex(oldKey);
        if (!result.exists) {
            throw new RuntimeException(String.format("非法替换key, oldKey: %s, newKey: %s, page: %s", oldKey, newKey, this));
        }
        // 需要保证该key替换后不影响顺序
        if (result.index > 0 && newKey.compareTo(entries[result.index - 1].key) <= 0) {
            throw new RuntimeException(String.format("非法替换key, oldKey: %s, newKey: %s, 新key替换后比前继key小, page: %s",
                    oldKey, newKey, this.toString()));
        }
        if (result.index < (keyCount - 1) && newKey.compareTo(entries[result.index + 1].key) >= 0) {
            throw new RuntimeException(String.format("非法替换key, oldKey: %s, newKey: %s, 新key替换后比后继key大, page: %s",
                    oldKey, newKey, this.toString()));
        }

        entries[result.index].key = newKey;// 换key
        // 更新页头已使用空间
        decreasePageUsedSize(oldKey);
        increasePageUsedSize(newKey);
        this.modified = true;// 标记页面发生了修改

        // 若替换的key为索引0, 递归触发minKey更改
        return result.index;
    }

    public boolean canInsert(String key) {
        // key槽未写满且页面空间充足
        return this.keyCount < Key_Max_Num && canWriteKey(key.getBytes(StandardCharsets.UTF_8));
    }

    public boolean canWriteKey(byte[] key) {
        // 页面空间充足
        return this.pageUsedSize + 4 + key.length + 8 <= Page_Size;
    }

    public boolean isIndexPage() {
        return pageType == Index_Page_Type;
    }

    public boolean isLeafPage() {
        return pageType == Leaf_Page_Type;
    }

    public String minKey() {
        if (keyCount <= 0) {
            throw new ArrayIndexOutOfBoundsException(String.format("当前页面keyCount为0, pageIdx: %d", pageIdx));
        }
        return entries[0].key;
    }

    public String maxKey() {
        if (keyCount <= 0) {
            throw new ArrayIndexOutOfBoundsException(String.format("当前页面keyCount为0, pageIdx: %d", pageIdx));
        }
        return entries[keyCount - 1].key;
    }

    public boolean hasPrePage() {
        return prePageIdx > Page.Invalid_PageIdx;
    }

    public boolean hasNextPage() {
        return nextPageIdx > Page.Invalid_PageIdx;
    }

    public void increasePageUsedSize(String key) {
        increasePageUsedSize(key.getBytes(StandardCharsets.UTF_8));
    }

    public void increasePageUsedSize(byte[] key) {
        pageUsedSize += (4 + key.length + 8);
    }

    public void decreasePageUsedSize(String key) {
        decreasePageUsedSize(key.getBytes(StandardCharsets.UTF_8));
    }

    public void decreasePageUsedSize(byte[] key) {
        pageUsedSize -= (4 + key.length + 8);
    }

    public ByteBuffer serialize() {
        ByteBuffer buffer = ByteBuffer.allocate(Page_Size);
        // 1.写页头
        buffer.putInt(pageType);
        buffer.putInt(pageIdx);
        buffer.putInt(parentPageIdx);
        buffer.putInt(keyCount);
        buffer.putInt(prePageIdx);
        buffer.putInt(nextPageIdx);
        buffer.putInt(pageUsedSize);

        // 2.写key和val
        for (int i = 0; i < keyCount; i++) {
            BEntry entry = entries[i];
            byte[] bytes = entry.key.getBytes(StandardCharsets.UTF_8);

            buffer.putInt(bytes.length);// key的长度
            buffer.put(bytes);
            buffer.putLong(entry.val);
        }
        if (buffer.position() != pageUsedSize) {
            throw new RuntimeException(String.format("pageIdx: %d的页面的pageUsedSize: %d 不等于序列化后的字节数: %d", pageIdx, pageUsedSize, buffer.position()));
        }
        // 3.填充空白字节
        if (pageUsedSize < Page_Size) {
            byte[] padding = new byte[Page_Size - pageUsedSize];
            buffer.put(padding);
        }
        buffer.flip();// 翻转待读
        return buffer;
    }

    public static Page deserialize(ByteBuffer buffer) {
        myAssert(buffer.remaining() == Page_Size, String.format("反序列化时buffer的可读字节: %d 不是页面大小: %d", buffer.remaining(), Page_Size));
        int startPos = buffer.position();
        // 1.读页头
        int pageType = buffer.getInt();
        int pageIdx = buffer.getInt();
        int parentPageIdx = buffer.getInt();
        int keyCount = buffer.getInt();
        int prePageIdx = buffer.getInt();
        int nextPageIdx = buffer.getInt();
        int pageUsedSize = buffer.getInt();

        BEntry[] entries = new BEntry[Key_Max_Num];
        // 2.读key和val
        for (int i = 0; i < keyCount; i++) {
            int keyLen = buffer.getInt();// 读key长度
            if (keyLen <= 0) {
                throw new RuntimeException(String.format("pageIdx: %d即将反序列化, i: %d, 非法keyLen: %d", pageIdx, i, keyLen));
            }
            byte[] bytes = new byte[keyLen];
            buffer.get(bytes, 0, keyLen);
            String key = new String(bytes, StandardCharsets.UTF_8);
            long val = buffer.getLong();
            entries[i] = new BEntry(key, val);
        }
        int endPos = buffer.position();

        myAssert(endPos - startPos == pageUsedSize, String.format("反序列后，buffer使用的字节数: %d 必须和页使用字节数: %d 一致, startPos: %d, endPos: %d",
                endPos - startPos, pageUsedSize, startPos, endPos));

        Page page = new Page(pageType, pageIdx, parentPageIdx);
        page.keyCount = keyCount;
        page.prePageIdx = prePageIdx;
        page.nextPageIdx = nextPageIdx;
        page.pageUsedSize = pageUsedSize;
        page.entries = entries;
        return page;
    }

    @Data
    public static class BEntry {
        public String key;
        public long val;// 可能是pageIdx或是offset

        public BEntry(String key, long val) {
            this.key = key;
            this.val = val;
        }
    }

    public static class SearchResult {
        public boolean exists;
        public int index;// key存在时为其索引，不存在时，index为key可能插入的位置

        public SearchResult(boolean exists, int index) {
            this.exists = exists;
            this.index = index;
        }
    }

    public String getPageHeaderString() {
        return "{pageIdx=" + pageIdx +
                ", pageType=" + pageType +
                ", parentPageIdx=" + parentPageIdx +
                ", keyCount=" + keyCount +
                ", prePageIdx=" + prePageIdx +
                ", nextPageIdx=" + nextPageIdx +
                ", pageUsedSize=" + pageUsedSize +
                '}';
    }

    public String getKeysString() {
        if (keyCount <= 0) {
            return "{}";
        }
        StringBuilder sb = new StringBuilder();
        sb.append('{');
        for (int i = 0; i < keyCount; i++) {
            sb.append(entries[i].key).append(',').append(' ');
        }
        sb.delete(sb.length() - 2, sb.length());
        sb.append('}');
        return sb.toString();
    }

    public String getKeyValsString() {
        if (keyCount <= 0) {
            return "{}";
        }
        StringBuilder sb = new StringBuilder();
        sb.append('{');
        for (int i = 0; i < keyCount; i++) {
            sb.append('{').append(entries[i].key).append(':').append(entries[i].val).append('}').append(", ");
        }
        sb.delete(sb.length() - 2, sb.length());
        sb.append('}');
        return sb.toString();
    }

    @Override
    public String toString() {
        return "page{" +
                "header=" + getPageHeaderString() +
                ", keys=" + getKeyValsString() +
                "}";
    }

    public void checkPage() {
        // 检查页面是否合法
        // 检查顺序排列: k1<k2
        for (int i = 1; i < keyCount; i++) {
            if (entries[i - 1].key.compareTo(entries[i].key) >= 0) {
                throw new RuntimeException(String.format("页内key顺序错误, pageIdx: %d, %d_key: %s, %d_key: %s", pageIdx, i - 1, entries[i - 1].key, i, entries[i].key));
            }
        }
        // 剩余的key槽都必须为null
        for (int i = keyCount; i < Key_Max_Num; i++) {
            if (entries[i] != null) {
                throw new RuntimeException(String.format("存在某个key槽不为空, pageIdx: %d, keyCount: %d, i: %d, key: %s", pageIdx, keyCount, i, entries[i].key));
            }
        }

        // 检查页头
        myAssert(pageType == Page.Index_Page_Type || pageType == Page.Leaf_Page_Type, String.format("非法pageType: %d", pageType));
        myAssert(pageUsedSize >= Page_Header_Meta_Size, String.format("非法pageUsedSize: %d", pageUsedSize));
    }

    public static void myAssert(boolean flag, String msg) {
        if (!flag) throw new RuntimeException(msg);
    }
}
